package bma.common.netty.client;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicReference;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.WriteCompletionEvent;

import bma.common.langutil.ai.stack.AIStack;
import bma.common.langutil.ai.stack.AIStackROOT;
import bma.common.netty.NettyUtil;
import bma.common.netty.handler.ChannelHandlerPlaceholder;

public class NettyClient {

	public static final String NETTY_CLIENT_INSTANCE = "netty_client_instance";

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(NettyClient.class);

	protected Channel channel;
	protected List<Object> listeners = new CopyOnWriteArrayList<Object>();
	protected boolean externalClose;
	protected AtomicReference<NettyClientSender> currentSender = new AtomicReference<NettyClientSender>();
	protected ConcurrentLinkedQueue<NettyClientSender> senderQueue = new ConcurrentLinkedQueue<NettyClientSender>();
	protected String placeholderName = NETTY_CLIENT_INSTANCE;

	protected NettyClient() {
		super();
	}

	public String getPlaceholderName() {
		return placeholderName;
	}

	public Channel getChannel() {
		return channel;
	}

	public List<Object> getListeners() {
		return new ArrayList<Object>(listeners);
	}

	public void setListeners(List<Object> list) {
		if (list != null) {
			for (Object lis : list) {
				addListener(lis);
			}
		}
	}

	public void addListener(Object lis) {
		if (lis == null)
			return;
		if (!this.listeners.contains(lis)) {
			boolean pass = false;
			if (lis instanceof NettyClientWriteCompleteListener) {
				pass = true;
			}
			if (lis instanceof NettyClientMessageListener) {
				pass = true;
			}
			if (lis instanceof NettyClientExceptionListener) {
				pass = true;
			}
			if (lis instanceof NettyClientCloseListener) {
				pass = true;
			}
			if (pass) {
				this.listeners.add(lis);
			} else {
				throw new ClassCastException("invalid listener type '"
						+ lis.getClass().getName() + "'");
			}
		}
	}

	public boolean removeListener(NettyClientMessageListener lis) {
		if (lis == null)
			return false;
		return this.listeners.remove(lis);
	}

	protected void handleFuture(AIStack<NettyClient> stack, ChannelFuture future) {
		if (future.isSuccess()) {
			stack.success(NettyClient.this);
		} else {
			stack.failure(future.getCause());
		}
	}

	public Integer getId() {
		if (this.channel == null)
			return null;
		return channel.getId();
	}

	public boolean isOpen() {
		if (channel == null)
			return false;
		return channel.isOpen();
	}

	public boolean isBound() {
		if (channel == null)
			return false;
		return channel.isBound();
	}

	public boolean isConnected() {
		if (channel == null)
			return false;
		return channel.isConnected();
	}

	public SocketAddress getLocalAddress() {
		if (channel == null)
			return null;
		return channel.getLocalAddress();
	}

	public SocketAddress getRemoteAddress() {
		if (channel == null)
			return null;
		return channel.getRemoteAddress();
	}

	public ChannelFuture write(Object message) {
		if (channel == null || !channel.isConnected()) {
			return null;
		}
		return channel.write(message);
	}

	public boolean write(final AIStack<NettyClient> stack, Object message) {
		if (channel == null || !channel.isConnected()) {
			return stack.failure(new IOException("channel not open"));
		}
		channel.write(message).addListener(new ChannelFutureListener() {

			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				handleFuture(stack, future);
			}
		});
		return false;
	}

	public ChannelFuture write(Object message, SocketAddress remoteAddress) {
		if (channel == null)
			return null;
		return channel.write(message, remoteAddress);
	}

	public boolean write(final AIStack<NettyClient> stack, Object message,
			SocketAddress remoteAddress) {
		if (channel == null || !channel.isConnected()) {
			return stack.failure(new IOException("channel not open"));
		}
		channel.write(message, remoteAddress).addListener(
				new ChannelFutureListener() {

					@Override
					public void operationComplete(ChannelFuture future)
							throws Exception {
						handleFuture(stack, future);
					}
				});
		return false;
	}

	public static class Receiver<TYPE> implements NettyClientMessageListener,
			NettyClientCloseListener, NettyClientExceptionListener {

		protected AIStack<TYPE> stack;
		protected boolean readOnce;

		public Receiver(AIStack<TYPE> stack, boolean readOnce) {
			super();
			this.stack = stack;
			this.readOnce = readOnce;
		}

		@Override
		public boolean exceptionCaught(NettyClient client, Throwable t) {
			stack.failure(t);
			return true;
		}

		@Override
		public void clientClose(NettyClient client, boolean remoteClose) {
			stack.failure(new NettyClientCloseException(
					remoteClose ? "remoteClose" : "close"));
		}

		@Override
		public boolean messageReceived(NettyClient client, Object message) {
			stack.success((TYPE) message);
			if (readOnce) {
				client.removeListener(this);
			}
			return true;
		}

	}

	public <TYPE> void receive(AIStack<TYPE> stack, boolean readOnce) {
		addListener(new Receiver<TYPE>(stack, readOnce));
	}

	public void close() {
		if (channel == null) {
			return;
		}
		if (channel.isOpen()) {
			externalClose = true;
			channel.close();
		}
	}

	public boolean close(final AIStack<NettyClient> stack) {
		if (channel == null) {
			return stack.success(this);
		}
		channel.close().addListener(new ChannelFutureListener() {

			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				handleFuture(stack, future);
			}
		});
		return false;
	}

	public ChannelFuture getCloseFuture() {
		if (channel == null)
			return null;
		return channel.getCloseFuture();
	}

	public boolean isReadable() {
		if (channel == null)
			return false;
		return channel.isReadable();
	}

	public boolean isWritable() {
		if (channel == null)
			return false;
		return channel.isWritable();
	}

	protected class CHNettyClientInstance extends SimpleChannelUpstreamHandler {

		public NettyClient client() {
			return NettyClient.this;
		}

		@Override
		public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
				throws Exception {
			NettyClientSender sender = currentSender.get();
			if (sender != null) {
				sender.clientClose(NettyClient.this, !externalClose);
				currentSender.set(null);
				for (NettyClientSender s : senderQueue) {
					s.clientClose(NettyClient.this, !externalClose);
				}
				senderQueue.clear();
			}
			for (Object lis : listeners) {
				try {
					if (lis instanceof NettyClientCloseListener) {
						NettyClientCloseListener l = (NettyClientCloseListener) lis;
						l.clientClose(NettyClient.this, !externalClose);
					}
				} catch (Exception err) {
				}
			}
			super.channelClosed(ctx, e);
		}

		@Override
		public void writeComplete(ChannelHandlerContext ctx,
				WriteCompletionEvent e) throws Exception {
			for (Object lis : listeners) {
				try {
					if (lis instanceof NettyClientWriteCompleteListener) {
						NettyClientWriteCompleteListener l = (NettyClientWriteCompleteListener) lis;
						l.writeComplete(NettyClient.this, e.getWrittenAmount());
					}
				} catch (Exception err) {
				}
			}
			super.writeComplete(ctx, e);
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
				throws Exception {
			boolean next = true;
			NettyClientSender sender = currentSender.get();
			if (sender != null) {
				if (sender.exceptionCaught(NettyClient.this, e.getCause())) {
					next = false;
				}
			}
			for (Object lis : listeners) {
				try {
					if (lis instanceof NettyClientExceptionListener) {
						NettyClientExceptionListener l = (NettyClientExceptionListener) lis;
						if (l.exceptionCaught(NettyClient.this, e.getCause())) {
							return;
						}
					}
				} catch (Exception err) {
				}
			}
			if (next) {
				super.exceptionCaught(ctx, e);
			}
		}

		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			boolean next = true;
			NettyClientSender sender = currentSender.get();
			if (sender != null) {
				if (sender.messageReceived(NettyClient.this, e.getMessage())) {
					next = false;
				}
			}
			for (Object lis : listeners) {
				try {
					if (lis instanceof NettyClientMessageListener) {
						NettyClientMessageListener l = (NettyClientMessageListener) lis;
						if (l.messageReceived(NettyClient.this, e.getMessage())) {
							return;
						}
					}
				} catch (Exception err) {
				}
			}
			if (next) {
				super.messageReceived(ctx, e);
			}
		}
	}

	protected ChannelHandler newHandler() {
		return new CHNettyClientInstance();
	}

	public static void addPlaceholder(ChannelPipeline p) {
		p.addLast(NETTY_CLIENT_INSTANCE, new ChannelHandlerPlaceholder());
	}

	@Override
	public String toString() {
		return channel == null ? super.toString() : channel.toString();
	}

	public void unbind() {
		ChannelPipeline p = channel.getPipeline();
		ChannelHandlerContext ctx = p.getContext(CHNettyClientInstance.class);
		if (ctx != null) {
			p.replace(ctx.getHandler(), ctx.getName(),
					new ChannelHandlerPlaceholder());
		}
	}

	public static NettyClient bind(Channel channel) {
		return bind(channel, NETTY_CLIENT_INSTANCE);
	}

	public static NettyClient bind(Channel channel, String handlerName) {
		ChannelPipeline p = channel.getPipeline();
		CHNettyClientInstance o = p.get(CHNettyClientInstance.class);
		if (o != null) {
			return o.client();
		}

		NettyClient nc = new NettyClient();
		nc.channel = channel;
		ChannelHandler ch = nc.newHandler();
		if (p.get(handlerName) != null) {
			p.replace(handlerName, handlerName, ch);
		} else if (p.get(NettyUtil.HANDLER_PLACEHOLDER) != null) {
			nc.placeholderName = NettyUtil.HANDLER_PLACEHOLDER;
			p.replace(NettyUtil.HANDLER_PLACEHOLDER,
					NettyUtil.HANDLER_PLACEHOLDER, ch);
		} else {
			p.addLast(handlerName, ch);
		}
		return nc;
	}

	public boolean missHandler(String name) {
		if (channel == null) {
			throw new NullPointerException("channel is null");
		}
		ChannelPipeline p = channel.getPipeline();
		return p.get(name) == null;
	}

	public void addHandler(String name, ChannelHandler ch) {
		if (channel == null) {
			throw new NullPointerException("channel is null");
		}
		ChannelPipeline p = channel.getPipeline();
		p.addBefore(placeholderName, name, ch);
	}

	public void removeHandler(String name) {
		if (channel == null) {
			throw new NullPointerException("channel is null");
		}
		ChannelPipeline p = channel.getPipeline();
		if (p.get(name) != null) {
			p.remove(name);
		}
	}

	public int getSendQueueSize() {
		return senderQueue.size();
	}

	public boolean isSendQueueEmpty() {
		return senderQueue.isEmpty();
	}

	public void send(NettyClientSender sender) {
		if (channel == null) {
			throw new NullPointerException("channel is null");
		}
		if (!isOpen()) {
			throw new NettyClientCloseException("closed");
		}
		if (log.isDebugEnabled()) {
			log.debug("{} send queue({})", this, sender);
		}
		senderQueue.add(sender);
		processSend();
	}

	protected void processSend() {
		if (currentSender.get() == null) {
			NettyClientSender s = this.senderQueue.peek();
			if (s == null) {
				if (log.isDebugEnabled()) {
					log.debug("{} process send end", this);
				}
				return;
			}
			if (!currentSender.compareAndSet(null, s)) {
				if (log.isDebugEnabled()) {
					log.debug("{} process send conflict,skip", this);
				}
				return;
			}
			this.senderQueue.remove(s);
			processSender(s);
		}
	}

	protected void processSender(final NettyClientSender s) {
		if (log.isDebugEnabled()) {
			log.debug("{} process send ({})", this, s);
		}
		AIStackROOT<Boolean> root = new AIStackROOT<Boolean>() {
			@Override
			public boolean end(Boolean result, Throwable t) {
				if (log.isDebugEnabled()) {
					log.debug("{} process send end ({})", NettyClient.this, s);
				}
				if (currentSender.compareAndSet(s, null)) {
					try {
						s.endSend(NettyClient.this);
					} catch (Exception e) {
					}
					processSend();
				}
				return true;
			}
		};
		s.beginSend(root, this);
	}

	public ChannelFutureListener closeListener() {
		return new ChannelFutureListener() {

			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				close();
			}
		};

	};
}
